package com.xiashitech.agent.grpc;

import com.xiashitech.agent.cache.data.CacheData;
import com.xiashitech.agent.cache.data.CacheSync;
import com.xiashitech.agent.config.StartConfig;
import com.xiashitech.agent.instrumentation.db.statement.sanitizer.impl.XSSqlSanitizer;
import com.xiashitech.agent.utils.JsonUtil;
import com.xiashitech.interfaces.apiservice.dto.AuthFactRequest;
import com.xiashitech.interfaces.apiservice.dto.AuthFactResponse;
import com.xiashitech.interfaces.grpc.RPCAuthFactRequest;
import com.xiashitech.interfaces.grpc.RPCAuthFactResponse;
import com.xiashitech.interfaces.grpc.RPCAuthFactServiceGrpc;
import io.grpc.ManagedChannel;
import io.grpc.stub.StreamObserver;

import java.util.concurrent.ConcurrentHashMap;

public class XsAuthFactClient {
    private final static XsAuthFactClient client = new XsAuthFactClient();
    private final ConcurrentHashMap<String, AuthFactRequest> requestHashMap = new ConcurrentHashMap<>();
    public static XsAuthFactClient getClient() {
        return client;
    }
    private RPCAuthFactServiceGrpc.RPCAuthFactServiceStub requestStub = null;

    public void supplementUserTracingGRPC(AuthFactRequest authFactRequest) {
        System.out.println(Thread.currentThread().getId() + ", supplementUserTracingGRPC enter, StartConfig.agentService: " + StartConfig.getParamValueFromKey(StartConfig.agentService));
        ManagedChannel channelRef = XsChannelManager.getManager().buildStubOfChannel("127.0.0.1", 9988);
        System.out.println(Thread.currentThread().getId() + ", supplementUserTracingGRPC, now channel identityHashCode: "+ System.identityHashCode(channelRef));
        System.out.println(Thread.currentThread().getId() + ", supplementUserTracingGRPC, now authFactRequest identityHashCode: "+ System.identityHashCode(authFactRequest));
        String requestKey = String.valueOf(authFactRequest.hashCode());
        requestHashMap.put(requestKey, authFactRequest);
        synchronized (client) {
            if(requestStub == null) {
                requestStub = RPCAuthFactServiceGrpc.newStub(channelRef);
            }
        }

        StreamObserver<RPCAuthFactResponse> responseStreamObserver = new StreamObserver<RPCAuthFactResponse>() {
            @Override
            public void onNext(RPCAuthFactResponse value) {
                String reqKey = value.getRequestKey();
                if (reqKey == null || "".equals(reqKey)) {
                    System.out.println(Thread.currentThread().getId() + ", reqkey is empty, the response is unavailable");
                    return;
                }
                // 采用这方式，每次调用，应该都会把外部的这个引用和当前 this 对象绑定到一起
                System.out.println(Thread.currentThread().getId() + ", " + System.identityHashCode(authFactRequest) + " : " + authFactRequest + " ===>的响应为===> " + value.getAuthFactResponseJson());
                // 将json 转成对象;
                AuthFactResponse authFactResponse = JsonUtil.convertStringToObject(value.getAuthFactResponseJson(), AuthFactResponse.class);
                // 获取 requestKey
                AuthFactRequest authFactRequestRef = requestHashMap.get(reqKey);
                if(authFactRequestRef == null) // 表示服务端的推送过来的 key 可能无意义 // TODO 是否会服务端预先推送？
                    return;
                CacheData cacheData = CacheSync.getInstance().getCacheData(authFactRequestRef);
                if (cacheData != null) {
                    cacheData.setReturnValue(authFactResponse);
                } else {
                    // FAILED，服务端推送过来的数据，客户端没有相应的缓存
                    System.out.println(Thread.currentThread().getId() + ", the cache ISNOT exist, should init");
                    CacheSync.getInstance().init(XSSqlSanitizer.class, "supplementUserTracingForJob", authFactRequestRef, authFactResponse);
                    new Thread(() -> {
                        supplementUserTracingGRPC(authFactRequestRef);
                    }).start();
                }
            }

            @Override
            public void onError(Throwable t) {
                System.out.println(Thread.currentThread().getId() + ", supplementUserTracingForJob response error");
            }

            @Override
            public void onCompleted() {
                System.out.println(Thread.currentThread().getId() + ", supplementUserTracingForJob response onCompleted");
            }
        };

        StreamObserver<RPCAuthFactRequest> requestStreamObserver = requestStub.biStreamAuthFact(responseStreamObserver);

        System.out.println(Thread.currentThread().getId() + ", requestStreamObserver begin post request ");
        // 将 requestStreamObserver 缓存起来，下次发送请求
        try {
            System.out.println(Thread.currentThread().getId() + ", authFactRequest: " + authFactRequest + " ====> " + JsonUtil.convertObjectToString(authFactRequest));
            RPCAuthFactRequest rpcAuthFactRequest = RPCAuthFactRequest.newBuilder()
                    .setRequestKey(requestKey) //
                    .setAuthFactRequestJson(JsonUtil.convertObjectToString(authFactRequest))
                    .build();
            System.out.println(Thread.currentThread().getId() + ", Build rpcAuthFactRequest finished: " + rpcAuthFactRequest);
            requestStreamObserver.onNext(rpcAuthFactRequest);
            requestStreamObserver.onCompleted();
        } catch (Throwable t) {
            t.printStackTrace();
        }
        System.out.println(Thread.currentThread().getId() + ", requestStreamObserver post finished ");
    }
}
